package com.safe.first.application.service;

import cn.hutool.core.lang.id.NanoId;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RStream;
import org.redisson.api.RedissonClient;
import org.redisson.api.StreamGroup;
import org.redisson.api.StreamMessageId;
import org.redisson.api.stream.StreamAddArgs;
import org.redisson.api.stream.StreamReadGroupArgs;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;

/**
 * @author Mickey
 * @since 2022/6/5
 **/
@Slf4j
@Component
@RequiredArgsConstructor
public class RedisQueueAppService {

    public static final String queue = "queued";
    public static final String queueGroup = "queue-my-group";
    public static final String queueConsumer = NanoId.randomNanoId(10);

    private final RedissonClient redissonClient;
    private final ThreadPoolTaskExecutor threadPoolTaskExecutor;
    private RStream<String, String> stream;

    public void add() {
        StreamAddArgs<String, String> entries = StreamAddArgs.entry("-1", new TestQueueObj(131L, "hello").toString());
        StreamMessageId add = stream.add(entries);
        log.info("redis stream add msg : {}", add);
    }

    @PostConstruct
    public void init() {
        stream = redissonClient.getStream(queue);
        if (!stream.isExists()) {
            log.info("redis stream create group");
            stream.createGroup(queueGroup, StreamMessageId.ALL);
            stream.createConsumer(queueGroup, queueConsumer);
        } else {
            List<StreamGroup> streamGroups = stream.listGroups();
            log.info("redis stream group num : {}", streamGroups.size());
            if (streamGroups.isEmpty()) {
                log.info("redis stream create group");
                stream.createGroup(queueGroup, StreamMessageId.ALL);
                stream.createConsumer(queueGroup, queueConsumer);
            }
        }

        new Thread(() -> {
            try {
                while (true) {
                    StreamReadGroupArgs streamReadArgs = StreamReadGroupArgs.greaterThan(StreamMessageId.NEVER_DELIVERED);
                    streamReadArgs.timeout(Duration.ZERO);
                    streamReadArgs.count(1);
                    Map<StreamMessageId, Map<String, String>> read = stream.readGroup(queueGroup, queueConsumer, streamReadArgs);
                    Set<Map.Entry<StreamMessageId, Map<String, String>>> entries = read.entrySet();
                    for (Map.Entry<StreamMessageId, Map<String, String>> entry : entries) {
                        StreamMessageId key = entry.getKey();
                        CompletableFuture.runAsync(() -> {
                            Map<String, String> value = entry.getValue();
                            log.info("redis msg handle : {}", value.get("-1"));
                        }, threadPoolTaskExecutor).thenAccept(res -> {
                            log.info("redis ack msgId  : {}", key);
                            stream.ack(queueGroup, key);
                            log.info("redis remove msg : {}", key);
                            long remove = stream.remove(key);
                            log.info("remove result    : {}", remove);
                        }).exceptionally(exception -> {
                            log.error("{}", exception.getMessage());
                            return null;
                        });
                    }
                }
            } catch (Exception e) {
                log.warn("{}", e.getMessage());
            }
        }).start();
    }

    @PreDestroy
    public void destroy() {
        RStream<Object, Object> stream = redissonClient.getStream(queue);
        stream.removeConsumer(queueGroup, queueConsumer);
        log.info("redis stream consumer remove");
    }

    @Data
    @AllArgsConstructor
    public static class TestQueueObj {
        private Long id;
        private String name;
    }

}
